package com.vehicle.common.redis.manager;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.vehicle.common.redis.RedisProperties;
import com.vehicle.common.redis.connection.RedisConnectionFactory;
import com.vehicle.common.util.Jsons;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import redis.clients.jedis.*;

import java.util.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * Created by yantingbin on 2017/7/1.
 */
public class RedisCacheManager implements CacheManager {
    private static final Logger logger = LoggerFactory.getLogger(RedisCacheManager.class);

    private RedisConnectionFactory factory;

    public RedisConnectionFactory getFactory() {
        return factory;
    }

    public void setFactory(RedisConnectionFactory factory) {
        this.factory = factory;
    }

    @Override
    public void init() {
        logger.info("begin init redis...");
        factory.init();
        test();
        logger.info("init redis success...");
    }

    private boolean isCluster() {
        return factory.isCluster();
    }

    private <R> R apply(Function<MultiKeyCommands, R> function, R d) {
        try (Jedis jedis = factory.getJedisConnection()) {
            return function.apply(jedis);
        } catch (Exception e) {
            logger.error("redis apply ex", e);
        }
        return d;
    }

    private <R> R applyCluster(Function<MultiKeyJedisClusterCommands, R> function, R d) {
        try (JedisCluster jedisCluste = factory.getClusterConnection()) {
            return function.apply(jedisCluste);
        } catch (Exception e) {
            logger.error("redis applyCluster ex", e);
        }
        return d;
    }

    private void apply(Consumer<MultiKeyCommands> function) {
        try(Jedis jedis=factory.getJedisConnection()) {
            function.accept(jedis);
        } catch (Exception e) {
            logger.error("redis apply ex", e);
        }
    }

    private void applyCluster(Consumer<MultiKeyJedisClusterCommands> function) {
        try(JedisCluster jedisCluster=factory.getClusterConnection()) {
            function.accept(jedisCluster);
        } catch (Exception e) {
            logger.error("redis applyCluster ex", e);
        }
    }

    private <R> R call(Function<JedisCommands, R> function, R d) {
        if (factory.isCluster()) {
            try(JedisCluster jedisCluster=factory.getClusterConnection()) {
                return function.apply(jedisCluster);
            } catch (Exception e) {
                logger.error("redis ex", e);
            }
        } else {
            try (Jedis jedis = factory.getJedisConnection()) {
                return function.apply(jedis);
            } catch (Exception e) {
                logger.error("redis ex", e);
            }
        }
        return d;
    }

    private void call(Consumer<JedisCommands> consumer) {
        if (factory.isCluster()) {
            try(JedisCluster jedisCluster=factory.getClusterConnection()) {
                consumer.accept(jedisCluster);

            } catch (Exception e) {
                logger.error("redis ex", e);
            }
        } else {
            try (Jedis jedis = factory.getJedisConnection()) {
                consumer.accept(jedis);
            } catch (Exception e) {
                logger.error("redis ex", e);
            }
        }
    }

    private void callCluster(Consumer<BinaryJedisCluster> consumer) {
        try(JedisCluster jedisCluster=factory.getClusterConnection()) {
            consumer.accept(jedisCluster);
        } catch (Exception e) {
            logger.error("redis ex", e);
        }
    }

    private void callSingle(Consumer<BinaryJedis> consumer) {
        try (Jedis jedis = factory.getJedisConnection()) {
            consumer.accept(jedis);
        } catch (Exception e) {
            logger.error("redis ex", e);
        }
    }

    private <R> R callCluster(Function<BinaryJedisCluster, R> function, R d) {
        try(JedisCluster jedisCluster= factory.getClusterConnection()) {
            return function.apply(jedisCluster);
        } catch (Exception e) {
            logger.error("redis ex", e);
        }
        return d;
    }

    private <R> R callSingle(Function<BinaryJedis, R> function, R d) {
        try(Jedis jedis= factory.getJedisConnection()) {
            return function.apply(jedis);
        } catch (Exception e) {
            logger.error("redis ex", e);
        }
        return d;
    }

    public long incr(String key) {
        return call(jedis -> jedis.incr(key), 0L);
    }

    public long incrBy(String key, long delt) {
        return call(jedis -> jedis.incrBy(key, delt), 0L);
    }

    @Override
    public void destroy() {
        if (factory != null) {
            factory.destroy();
        }
    }

    @Override
    public <T> T get(String key, Class<T> tClass) {
        String value = call(jedis -> jedis.get(key), null);
        if (value == null) {
            return null;
        }
        if (tClass == String.class) {
            return (T) value;
        }
        return Jsons.fromJson(value, tClass);
    }

    @Override
    public byte[] get(byte[] key) {
        byte[] value = null;
        if (factory.isCluster()) {
            value = callCluster(binaryJedisCluster -> binaryJedisCluster.get(key), null);
        } else {
            value = callSingle(binaryJedisCluster -> binaryJedisCluster.get(key), null);
        }
        return value;
    }

    public Object get(String key) {
        String value = call(jedis -> jedis.get(key), null);
        if (value == null) {
            return null;
        }
        return Jsons.fromJson(value);
    }

    @Override
    public void set(String key, String value) {
        set(key, value, 0);
    }

    @Override
    public void set(String key, Object value) {
        set(key, value, 0);
    }

    @Override
    public void set(String key, Object value, int expireTime) {
        set(key, Jsons.toJson(value), expireTime);
    }

    @Override
    public void set(byte[] key, byte[] value) {
        set(key, value, 0);
    }

    @Override
    public void set(byte[] key, byte[] value, int expireTime) {
        if (factory.isCluster()) {
            callCluster(jedis -> {
                jedis.setex(key, expireTime, value);
            });
        } else {
            callSingle(binaryJedis -> binaryJedis.setex(key, expireTime, value));
        }
    }

    public void set(String key, String value, int time) {
        call(jedis -> {
            jedis.set(key, value);
            if (time > 0) {
                jedis.expire(key, time);
            }
        });
    }

    @Override
    public void del(String key) {
        call(jedis -> {
            jedis.del(key);
        });
    }


    @Override
    public void hset(String key, String field, String value) {
        call(jedis -> jedis.hset(key, field, value));
    }

    @Override
    public void hset(String key, String field, Object value) {
        call(jedis -> jedis.hset(key, field, Jsons.toJson(value)));
    }

    @Override
    public <T> T hget(String key, String field, Class<T> tClass) {
        String value = call(jedis -> {
            return jedis.hget(key, field);
        }, null);
        if (value == null) {
            return null;
        }
        if (tClass == String.class) {
            return (T) value;
        }
        return Jsons.fromJson(value, tClass);
    }

    @Override
    public void hdel(String key, String field) {
        call(jedis -> jedis.hdel(key, field));
    }

    public Map<String, String> hgetAll(String key) {
        return call(jedis -> jedis.hgetAll(key), Collections.<String, String>emptyMap());
    }

    @Override
    public <T> Map<String, T> hgetAll(String key, Class<T> clazz) {
        Map<String, String> result = hgetAll(key);
        if (result.isEmpty()) {
            return Collections.emptyMap();
        }
        Map<String, T> newMap = new HashMap(result.size());

        result.forEach((k, v) -> newMap.put(k, Jsons.fromJson(v, clazz)));

        return newMap;
    }

    @Override
    public long hincrBy(String key, String field, long value) {
        return call(jedis -> jedis.hincrBy(key, field, value), 0L);
    }

    @Override
    public Set<String> hkeys(String key) {
        return call(jedis -> jedis.hkeys(key), Collections.<String>emptySet());
    }

    @Override
    public <T> List<T> hmget(String key, Class<T> clazz, String... fields) {
        return call(jedis -> jedis.hmget(key, fields), Collections.<String>emptyList())
                .stream()
                .map(s -> Jsons.fromJson(s, clazz)).collect(Collectors.toList());
    }

    @Override
    public void hmset(String key, Map<String, String> hash, int time) {
        call(jedis -> {
            jedis.hmset(key, hash);
            if (time > 0) {
                jedis.expire(key, time);
            }
        });
    }

    @Override
    public void hmset(String key, Map<String, String> hash) {
        hmset(key, hash, 0);
    }

    @Override
    public void zAdd(String key, String value) {
        call(jedis -> jedis.zadd(key, 0, value));
    }

    @Override
    public Long zCard(String key) {
        return call(jedis -> jedis.zcard(key), 0L);
    }

    @Override
    public void zRem(String key, String value) {
        call(jedis -> jedis.zrem(key, value));
    }

    @Override
    public <T> List<T> zrange(String key, int start, int end, Class<T> clazz) {
        Set<String> value = call(jedis -> jedis.zrange(key, start, end), null);
        return toList(value, clazz);
    }

    @Override
    public void lpush(String key, String value) {
        call(jedis -> jedis.lpush(key, value));
    }

    @Override
    public void lpush(String key, Object value) {
        call(jedis -> jedis.lpush(key, Jsons.toJson(value)));
    }

    @Override
    public void rpush(String key, String value) {
        call(jedis -> jedis.rpush(key, value));
    }

    @Override
    public void rpush(String key, Object value) {
        call(jedis -> jedis.rpush(key, Jsons.toJson(value)));
    }

    @Override
    public <T> T lpop(String key, Class<T> clazz) {
        String value = call(jedis -> jedis.lpop(key), null);
        if (value == null) {
            return null;
        }
        if (clazz == String.class) {
            return (T) value;
        }
        return Jsons.fromJson(value, clazz);
    }

    @Override
    public <T> T rpop(String key, Class<T> clazz) {
        String value = call(jedis -> jedis.rpop(key), null);
        if (value == null) {
            return null;
        }
        if (clazz == String.class) {
            return (T) value;
        }
        return Jsons.fromJson(value, clazz);
    }

    @Override
    public <T> List<T> lrange(String key, int start, int end, Class<T> clazz) {
        return call(jedis -> jedis.lrange(key, start, end), Collections.<String>emptyList()).stream().map(s -> Jsons.fromJson(s, clazz)).collect(Collectors.toList());
    }

    @Override
    public long llen(String key) {
        return call(jedis -> jedis.llen(key), 0L);
    }

    @Override
    public void lRem(String key, String value) {
        call(jedis -> jedis.lrem(key, 0L, value));
    }

    @Override
    public void aAdd(String key, String value) {
        call(jedis -> jedis.sadd(key, value));
    }

    @Override
    public long sCard(String key) {
        return call(jedis -> jedis.scard(key), 0L);
    }

    @Override
    public void sRem(String key, String value) {
        call(jedis -> jedis.srem(key, value));
    }

    @Override
    public <T> List<T> sScan(String key, Class<T> clazz, int start) {
        List<String> list = call(jedis -> jedis.sscan(key, Integer.toString(start), new ScanParams().count(10)).getResult(), null);
        return toList(list, clazz);
    }

    private <T> List<T> toList(Collection<String> value, Class<T> clazz) {
        if (!value.isEmpty()) {
            if (clazz == String.class) {
                return (List<T>) new ArrayList<>(value);
            }
            return value.stream().map(s -> Jsons.fromJson(s, clazz)).collect(Collectors.toList());
        } else {
            return null;
        }
    }

    public void test() {
        if (factory.isCluster()) {
            JedisCluster cluster = factory.getClusterConnection();
            if (cluster == null) throw new RuntimeException("init redis cluster error.");
        } else {
            Jedis jedis = factory.getJedisConnection();
            if (jedis == null) throw new RuntimeException("init redis error, can not get connection.");
            jedis.close();
        }
    }

    public List<String> getAllKeys(String key) {
        List<String> result = Lists.newArrayList();

        ScanParams scanParams = new ScanParams().count(100).match(key);

        String cur = redis.clients.jedis.ScanParams.SCAN_POINTER_START;

        boolean cycleIsFinished = false;

        while (!cycleIsFinished) {

            final String curSub = cur;

            ScanResult<String> scanResult = null;

            if (isCluster()) {
                scanResult = applyCluster(jedis -> jedis.scan(curSub, scanParams), null);
            } else {
                scanResult = apply(jedis -> jedis.scan(curSub, scanParams), null);
            }

            //do whatever with the key-value pairs in result

            List<String> list = scanResult.getResult();

            result.addAll(list);

            cur = scanResult.getStringCursor();

            if (cur.equals("0")) {
                cycleIsFinished = true;
            }
        }

        return result;
    }
}
